package com.mo.axion.stream;

import com.mo.axion.common.constant.AxionConstant;
import com.mo.axion.common.domain.AnchorInfo;
import com.mo.axion.stream.constants.AxionFlinkConstant;
import com.mo.axion.stream.function.AxionFlatMapFunction;
import com.mo.axion.stream.sink.AxionSink;
import com.mo.axion.stream.utils.MessageQueueUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Mo
 * @createTime 2024/4/13 17:07
 * @description Flink启动类
 */
@Slf4j
public class AxionBootStrap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        
        //  1. 获取KafkaConsumer
        KafkaSource<String> kafkaConsumer = MessageQueueUtils.getKafkaConsumer(AxionFlinkConstant.TOPIC_NAME, AxionFlinkConstant.GROUP_ID, AxionFlinkConstant.BROKER);
        DataStreamSource<String> kafkaSource = environment.fromSource(kafkaConsumer, WatermarkStrategy.noWatermarks(), AxionFlinkConstant.SOURCE_NAME);
        System.out.println("Got the KafkaConsumer now!!");

        //  2. 数据转换处理
        SingleOutputStreamOperator<AnchorInfo> dataStream = kafkaSource.flatMap(new AxionFlatMapFunction()).name(AxionFlinkConstant.FUNCTION_NAME);

        //  3. 将实时数据多维度写入Redis
        dataStream.addSink(new AxionSink()).name(AxionFlinkConstant.SINK_NAME);
        environment.execute(AxionFlinkConstant.JOB_NAME);
    }
}
